package com.stars.easyms.mq.core.rocketmq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.stars.easyms.mq.asyn.AsynReceiveMQMessage;
import com.stars.easyms.mq.dto.MQInfoDTO;
import com.stars.easyms.mq.enums.DestinationType;
import com.stars.easyms.mq.properties.EasyMsMQProperties;
import com.stars.easyms.base.util.GenericTypeUtil;
import com.stars.easyms.mq.constant.MQConstants;
import com.stars.easyms.mq.core.MQListener;
import com.stars.easyms.mq.dto.MQMessageDTO;
import com.stars.easyms.mq.enums.MQType;
import com.stars.easyms.mq.service.MQMessageService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;

/**
 * RocketMQ监听类
 *
 * @author guoguifang
 * @date 2018-04-23 13:54
 * @since 1.0.0
 */
public final class RocketMQListener implements MQListener {

    private static final Logger logger = LoggerFactory.getLogger(RocketMQListener.class);
    private DefaultMQPushConsumer consumer;
    private String nameServer;
    private String groupName;
    private MQInfoDTO mqInfoDTO;
    private String topic;
    private MQMessageService mqMessageService;
    private AsynReceiveMQMessage asynReceiveMQMessage;
    private String applicationId;
    private volatile boolean running;

    @Override
    public synchronized void start() {
        if (this.running) {
            logger.error("The MQ listener({}) is already running!", topic);
            return;
        }

        consumer = new DefaultMQPushConsumer(groupName + topic);
        consumer.setNamesrvAddr(nameServer);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setConsumeMessageBatchMaxSize(10);
        try {
            consumer.subscribe(topic, "*");
        } catch (MQClientException e) {
            logger.error("The MQ listener(" + this.topic + ") subscribe failure!", e);
        }
        consumer.registerMessageListener((List<MessageExt> msgs, ConsumeConcurrentlyContext context) -> {
            List<MQMessageDTO> insertMQMessageDTOList = new ArrayList< >();
            for (MessageExt msg : msgs) {
                List<MQMessageDTO> mqMessageDTOList = null;
                Object messageObj = null;
                String text = new String(msg.getBody(), StandardCharsets.UTF_8);
                // 首先转换成MQMessageDTO，如果转换后messageId不为空则表示是内部mq调用
                try {
                    mqMessageDTOList = JSON.parseObject(text, new TypeReference<List<MQMessageDTO>>() {
                    });
                    messageObj = mqMessageDTOList;
                } catch (Exception e) {
                    logger.debug("消息转换List<MQMessageDTO>失败!", e);
                }
                try {
                    if (mqMessageDTOList == null) {
                        Class<?> parameterClass = this.mqInfoDTO.getParameterClass();
                        if (String.class.isAssignableFrom(parameterClass)) {
                            messageObj = text;
                        } else {
                            Type parameterType = this.mqInfoDTO.getParameterType();
                            if (!(parameterType instanceof ParameterizedType)) {
                                messageObj = JSON.parseObject(text, parameterClass);
                            } else {
                                messageObj = GenericTypeUtil.parseObject(text, (ParameterizedType) parameterType, parameterClass);
                            }
                        }
                    }
                    String messageStr = messageObj instanceof String ? (String) messageObj : JSON.toJSONString(messageObj);
                    // 记录待接收处理消息日志
                    logger.info("接收到目标({})待处理MQ消息,消息内容:{}", this.topic, messageStr);
                    if (mqMessageDTOList != null) {
                        for (MQMessageDTO mqMessageDTO : mqMessageDTOList) {
                            mqMessageDTO.setId(mqMessageService.getId());
                            mqMessageDTO.setType(MQType.RECEIVE.getCode());
                            mqMessageDTO.setDestinationType(DestinationType.QUEUE.getCode());
                            mqMessageDTO.setDestSys(applicationId);
                            mqMessageDTO.setStatus(MQConstants.MQ_STATUS_PROCESSING);
                            mqMessageDTO.setFailCount(0L);
                            mqMessageDTO.setRedeliveryPolicy(RocketMQListener.this.mqInfoDTO.getRedeliveryPolicy().getCode());
                            mqMessageDTO.setRedeliveryDelay(RocketMQListener.this.mqInfoDTO.getRedeliveryDelay());
                            mqMessageDTO.setFireTime(new Timestamp(System.currentTimeMillis()));
                            insertMQMessageDTOList.add(mqMessageDTO);
                        }
                    } else {
                        MQMessageDTO mqMessageDTO = new MQMessageDTO();
                        mqMessageDTO.setId(mqMessageService.getId());
                        mqMessageDTO.setMessageId(mqMessageService.getMessageId());
                        mqMessageDTO.setKey(RocketMQListener.this.topic);
                        mqMessageDTO.setMessage(messageStr);
                        mqMessageDTO.setType(MQType.RECEIVE.getCode());
                        mqMessageDTO.setDestinationType(DestinationType.QUEUE.getCode());
                        mqMessageDTO.setDestSys(applicationId);
                        mqMessageDTO.setStatus(MQConstants.MQ_STATUS_PROCESSING);
                        mqMessageDTO.setFailCount(0L);
                        mqMessageDTO.setRedeliveryPolicy(RocketMQListener.this.mqInfoDTO.getRedeliveryPolicy().getCode());
                        mqMessageDTO.setRedeliveryDelay(RocketMQListener.this.mqInfoDTO.getRedeliveryDelay());
                        mqMessageDTO.setFireTime(new Timestamp(System.currentTimeMillis()));
                        insertMQMessageDTOList.add(mqMessageDTO);
                    }
                } catch (Exception e) {
                    logger.error("接收目标(" + this.topic + ")MQ消息失败,消息内容:" + this.consumer.toString() + "!", e);
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
            if (!insertMQMessageDTOList.isEmpty()) {
                for (MQMessageDTO mqMessageDTO : insertMQMessageDTOList) {
                    asynReceiveMQMessage.add(mqMessageDTO);
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        try {
            this.consumer.start();
            this.running = true;
            logger.info("The MQ listener({}) start up success!", this.topic);
        } catch (MQClientException e) {
            logger.error("The MQ listener(" + this.topic + ") start up failure!", e);
        }
    }

    @Override
    public synchronized void stop() {
        if (!this.running) {
            logger.error("The MQ listener({}) is not started!", topic);
            return;
        }
        try {
            this.consumer.shutdown();
            this.running = false;
            logger.info("The MQ listener({}) shut down success!", topic);
        } catch (Exception e) {
            logger.error("The MQ listener(" + topic + ") shut down failure!", e);
        }
    }

    @Override
    public synchronized boolean isRunning() {
        return this.running;
    }

    public RocketMQListener(String applicationId, EasyMsMQProperties easyMsMQProperties, MQMessageService mqMessageService, AsynReceiveMQMessage asynReceiveMQMessage, MQInfoDTO mqInfoDTO, String topic) {
        this.mqMessageService = mqMessageService;
        this.asynReceiveMQMessage = asynReceiveMQMessage;
        this.mqInfoDTO = mqInfoDTO;
        this.applicationId = applicationId;
        this.nameServer = easyMsMQProperties.getRocketmq().getNameServer();
        this.groupName = this.applicationId + "_CONSUMER_";
        this.topic = topic;
    }
}
